package com.rocsea.robot.service;

import com.alibaba.fastjson.JSONObject;
import com.rocsea.robot.builder.DefaultConcreteKettleBuilder;
import com.rocsea.robot.builder.KettleBuildRobot;
import com.rocsea.robot.config.build.KettleBuildConfig;
import com.rocsea.robot.config.jdbc.common.JdbcConfig;
import com.rocsea.robot.constant.EtlConstant;
import com.rocsea.robot.exception.BusinessException;
import com.rocsea.robot.factory.KettleBuildDatabaseFactory;
import com.rocsea.robot.model.bo.EtlExcelColumn;
import com.rocsea.robot.model.bo.EtlExcelSheet;
import com.rocsea.robot.model.bo.KettleProduct;
import com.rocsea.robot.model.bo.TableCheckConfig;
import com.rocsea.robot.model.entity.DmlSubscribeConfigDO;
import com.rocsea.robot.model.enums.*;
import com.rocsea.robot.model.request.EtlSyncCheckConfigBuildRequest;
import com.rocsea.robot.model.request.KettleExportRequest;
import com.rocsea.robot.service.http.BackendToolService;
import com.rocsea.robot.template.Template;
import com.rocsea.robot.utils.*;
import com.rocsea.robot.model.entity.JobDO;
import com.rocsea.robot.model.entity.TransformationDO;
import com.rocsea.robot.factory.KettleTemplateFactory;
import com.rocsea.robot.model.request.KettleBuildRequest;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * kettle机器服务
 *
 * @Author RocSea
 * @Date 2022/12/2
 */
@Service
public class KettleRobotService {
    private static final Log logger = LogFactory.getLog(KettleRobotService.class);
    @Resource
    private TransformationService transformationService;

    @Resource
    private JobService jobService;

    @Resource
    private DirectoryService directoryService;

    @Resource
    private DatabaseService databaseService;

    @Resource
    private KettleExtService kettleExtService;

    @Resource
    private KettleBuildConfig kettleBuildConfig;

    @Resource
    private BackendToolService backendToolService;

    /**
     * 自动构建
     *
     * @param request 请求
     * @throws Exception 异常
     */
    public void autoBuild(KettleBuildRequest request) throws Exception {
        // check request param
        checkRequestParam(request);

        // select build strategy
        String name = KettleBuildStrategyEnum.geText(request.getStrategyType());
        Template template = KettleTemplateFactory.getTemplate(name);

        // make product
        KettleBuildProxy proxy = new KettleBuildProxy(transformationService, jobService,
                directoryService, databaseService, kettleBuildConfig);
        KettleBuildRobot kettleBuildRobot = new KettleBuildRobot(new DefaultConcreteKettleBuilder(template, proxy));
        KettleProduct kettleProduct = kettleBuildRobot.makeProduct(request);

        // write file
        String path = kettleBuildConfig.getPath() + EtlConstant.ETL_KETTLE_PATH + request.getSourceTableName() + ".sql";
        if (!FileUtils.writeTextToFile(path, kettleProduct.get())) {
            throw new BusinessException("kettle语句写入文件失败");
        }
    }

    private void checkRequestParam(KettleBuildRequest request) {
        KettleBuildModuleEnum.geText(request.getModuleType());
        KettleBuildStrategyEnum.geText(request.getStrategyType());
    }

    /**
     * 构建删除
     *
     * @param idDirectory 目录ID
     * @return String
     */
    public String buildDel(Long idDirectory) {
        String transformationRelatedDelete = buildTransformationRelatedDelete(idDirectory);
        String jobRelatedDelete = buildJobRelatedDelete(idDirectory);
        return transformationRelatedDelete + jobRelatedDelete;
    }

    private String buildJobRelatedDelete(Long idDirectory) {
        StringBuilder sql = new StringBuilder();
        List<JobDO> jobDOS = jobService.listJob(idDirectory);
        String jobIds = jobDOS.stream().map(JobDO::getIdJob).map(String::valueOf).collect(Collectors.joining(","));
        sql.append(MysqlBuilderUtils.buildJobDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobAttributeDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobHopDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobLockDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobEntryDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobEntryAttributeDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobEntryCopyDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobNoteDeleteIn(jobIds));
        sql.append(MysqlBuilderUtils.buildJobEntryDatabaseDeleteIn(jobIds));
        return sql.toString();
    }

    private String buildTransformationRelatedDelete(Long idDirectory) {
        StringBuilder sql = new StringBuilder();
        List<TransformationDO> transformationDOS = transformationService.listTransformation(idDirectory);
        String transformationIds = StringUtils.join(transformationDOS.stream()
                .map(TransformationDO::getIdTransformation).toArray(), ",");
        sql.append(MysqlBuilderUtils.buildTransformationDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildTransAttributeDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildStepDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildStepAttributeDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildStepDatabaseDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildTransHopDeleteIn(transformationIds));
        sql.append(MysqlBuilderUtils.buildTranStepConditionDeleteIn(transformationIds));
        return sql.toString();
    }

    public void export(KettleExportRequest request, HttpServletResponse response) throws IOException {
        StringBuilder sql = new StringBuilder();
        Long idDirectory = directoryService.export(request, sql);
        transformationService.export(idDirectory, sql);
        jobService.export(idDirectory, sql);
        // download
        String fullName = DateFormatUtils.format(new Date(), "MMdd") + "_" + request.getSourceTableName() + "(导出).sql";
        FileUtils.download(response, sql.toString(), fullName);

    }

    public void fullAuto(MultipartFile file, String range) throws Exception {
        List<EtlExcelSheet> etlExcelSheets = EtlExcelUtils.readExcel(file.getInputStream(), range);
        autoBuildCreateTable(etlExcelSheets);
        autoBuildKettle(etlExcelSheets);
        autoBuildCheckConfig(etlExcelSheets);
        autoBuildSubscribeConfig(etlExcelSheets);
    }

    private void autoBuildSubscribeConfig(List<EtlExcelSheet> etlExcelSheets) {
        StringBuilder sql = new StringBuilder();
        for (EtlExcelSheet etlExcelSheet : etlExcelSheets) {
            if (etlExcelSheet.getHasPhysicalDel()) {
                DmlSubscribeConfigDO dmlSubscribeConfigDO = new DmlSubscribeConfigDO();
                dmlSubscribeConfigDO.setTargetServiceModule(ServiceModuleEnum.SELL_OUT.getValue());
                dmlSubscribeConfigDO.setSourceTableTopic(etlExcelSheet.getSourceSchemaName() + "_" + etlExcelSheet.getSourceTableName());
                dmlSubscribeConfigDO.setTargetSchemaName(EtlConstant.DEFAULT_SELL_DATABASE);
                dmlSubscribeConfigDO.setTargetTableName(etlExcelSheet.getTargetTableName());
                dmlSubscribeConfigDO.setOperationType(DmlTypeEnum.DELETE.getValue());
                dmlSubscribeConfigDO.setEnableStatus(EnableEnum.ENABLE.getValue());
                dmlSubscribeConfigDO.setDmlConsumeType(ConsumeTypeEnum.CONSUME_TYPE_ONE_TO_ONE.getValue());
                sql.append(MysqlExtBuilderUtils.buildDmlSubscribeConfig(dmlSubscribeConfigDO));
            }
        }
        if (StringUtils.isEmpty(sql.toString())) {
            return;
        }
        String path = kettleBuildConfig.getPath() + EtlConstant.ETL_CONFIG_PATH + "dml_config.sql";
        if (!FileUtils.writeTextToFile(path, sql.toString())) {
            throw new BusinessException("物理删除订阅配置语句写入文件失败");
        }
    }

    private void autoBuildCheckConfig(List<EtlExcelSheet> etlExcelSheetList) {
        StringBuilder sql = new StringBuilder();
        Map<String, List<EtlExcelSheet>> listMap = etlExcelSheetList.stream().collect(Collectors.groupingBy(EtlExcelSheet::getSourceSchemaName));
        for (List<EtlExcelSheet> excelSheets : listMap.values()) {
            final String sourceSchemaName = excelSheets.get(0).getSourceSchemaName();
            final String serviceModule = excelSheets.get(0).getServiceModule();
            EtlSyncCheckConfigBuildRequest request = new EtlSyncCheckConfigBuildRequest();
            request.setSourceServiceModule(KettleTranslateUtils.getServiceModuleByExcel(serviceModule));
            request.setSourceSchemaName(sourceSchemaName);
            List<TableCheckConfig> tableCheckConfigs = getTableCheckConfigList(excelSheets);
            request.setTableCheckConfigList(tableCheckConfigs);
            sql.append(kettleExtService.autoBuild(request));
        }
        String path = kettleBuildConfig.getPath() + EtlConstant.ETL_CONFIG_PATH + "check_config.sql";
        if (!FileUtils.writeTextToFile(path, sql.toString())) {
            throw new BusinessException("检查配置语句写入文件失败");
        }
    }

    private List<TableCheckConfig> getTableCheckConfigList(List<EtlExcelSheet> excelSheets) {
        List<TableCheckConfig> tableCheckConfigList = new ArrayList<>();
        for (EtlExcelSheet excelSheet : excelSheets) {
            TableCheckConfig tableCheckConfig = new TableCheckConfig();
            tableCheckConfig.setSourceTableName(excelSheet.getSourceTableName());
            tableCheckConfig.setTargetTableName(excelSheet.getTargetTableName());
            boolean physicalDelete = excelSheet.getHasPhysicalDel();
            tableCheckConfig.setSourceDeleteType(KettleTranslateUtils.getDeleteType(physicalDelete));
            if (physicalDelete) {
                String sourceWhereCondition = EtlConstant.DEFAULT_SOURCE_WHERE_CONDITION;
                if (excelSheet.getHasLogicDel()) {
                    sourceWhereCondition = EtlConstant.DEFAULT2_SOURCE_WHERE_CONDITION;
                    sourceWhereCondition.replace("deleted", excelSheet.getSourceColumnName("deleted"));
                }
                sourceWhereCondition.replace("update_time", excelSheet.getSourceColumnName("deleted"));
                tableCheckConfig.setSourceWhereCondition(sourceWhereCondition);
            }
            tableCheckConfigList.add(tableCheckConfig);
        }
        return tableCheckConfigList;
    }

    private void autoBuildKettle(List<EtlExcelSheet> etlExcelSheets) throws Exception {
        for (EtlExcelSheet etlExcelSheet : etlExcelSheets) {
            KettleBuildRequest kettleBuildRequest = new KettleBuildRequest();
            kettleBuildRequest.setModuleType(KettleTranslateUtils.getBuildModuleValue(etlExcelSheet.getServiceModule()));
            kettleBuildRequest.setSourceSchemaName(etlExcelSheet.getSourceSchemaName());
            kettleBuildRequest.setSourceTableName(etlExcelSheet.getSourceTableName());
            kettleBuildRequest.setTargetTableName(etlExcelSheet.getTargetTableName());
            kettleBuildRequest.setStrategyType(KettleBuildStrategyEnum.PAGING_UPDATE.getValue());
            List<EtlExcelColumn> excludeColumnList = etlExcelSheet.getEtlExcelColumnList().stream()
                    .filter(x -> "否".equals(x.getHasReserve())).collect(Collectors.toList());
            List<EtlExcelColumn> includeColumnList = etlExcelSheet.getEtlExcelColumnList().stream()
                    .filter(x -> "是".equals(x.getHasReserve())).collect(Collectors.toList());
            String sourceExcludeColumns = excludeColumnList.stream().map(EtlExcelColumn::getSourceColumnEnName).collect(Collectors.joining(","));
            kettleBuildRequest.setSourceExcludeColumns(sourceExcludeColumns);
            Map<String, String> aliasMapping = new HashMap<>();
            List<EtlExcelColumn> needAliasMappingColumns = includeColumnList.stream()
                    .filter(x -> !x.getSourceColumnEnName().equals(x.getSuggestedEnName())).collect(Collectors.toList());
            for (EtlExcelColumn column : needAliasMappingColumns) {
                aliasMapping.put(column.getSourceColumnEnName(), column.getSuggestedEnName());
            }
            kettleBuildRequest.setAliasMapping(aliasMapping);
            logger.info(JSONObject.toJSON(kettleBuildRequest));
            autoBuild(kettleBuildRequest);
        }
    }

    private void autoBuildCreateTable(List<EtlExcelSheet> etlExcelSheets) {
        for (EtlExcelSheet etlExcelSheet : etlExcelSheets) {
            final String serviceModule = etlExcelSheet.getServiceModule();
            final String schemaName = etlExcelSheet.getSourceSchemaName();
            final String tableName = etlExcelSheet.getSourceTableName();
            final String buildModule = KettleTranslateUtils.getBuildModuleText(serviceModule);
            JdbcConfig database = KettleBuildDatabaseFactory.getDatabase(buildModule);
            String sourceCreateTableSql = DatabaseUtils.getCreateTableSql(database.getUrl(), database.getUsername(),
                    database.getPassword(), schemaName, tableName);
            String mysqlTableSql = EtlStandardUtils.getCreateTableSqlByExcel(sourceCreateTableSql, etlExcelSheet);
            String mysqlPath = kettleBuildConfig.getPath() + EtlConstant.ETL_TABLE_PATH + etlExcelSheet.getSourceTableName() + "(mysql).sql";
            if (!FileUtils.writeTextToFile(mysqlPath, mysqlTableSql)) {
                throw new BusinessException("mysql建表语句写入文件失败");
            }
            String oracleTableSql = backendToolService.replace(SqlReplaceDbTypeEnum.ORACLE, mysqlTableSql);
            String oraclePath = kettleBuildConfig.getPath() + EtlConstant.ETL_TABLE_PATH + etlExcelSheet.getSourceTableName() + "(oracle).sql";
            if (!FileUtils.writeTextToFile(oraclePath, oracleTableSql)) {
                throw new BusinessException("oracle建表语句写入文件失败");
            }
            String sqlserverCreateTableSql = MysqlToSqlserver.generateSqlserverCreateTableSql(mysqlTableSql);
            String formatSqlserverSql = kettleExtService.formatMss(sqlserverCreateTableSql);

            String sqlserverPath = kettleBuildConfig.getPath() + EtlConstant.ETL_TABLE_PATH + etlExcelSheet.getSourceTableName() +
                    "(sqlserver).sql";
            if (!FileUtils.writeTextToFile(sqlserverPath, formatSqlserverSql)) {
                throw new BusinessException("sqlserver建表语句写入文件失败");
            }
        }
    }


}
